"""
Case Type   : 极致RTO支持备机可读
Case Name   : 资源池化网络复制双集群，集群间进行切换，新备集群开启极致RTO，switchover场景下验证备集群可读
Create At   : 2025/04
Owner       : haomeng
Description :
    1、原主备集群进行集群间切换，查看切换后的集群状态
    2、新备集群内开启极致RTO特性，重启使其生效
    3、新主集群内创建行存段页式range分区表，插入数据
    4、新备集群首备节点查看表数据
    5、新备集群从备节点查看表数据
    6、新备集群内进行switchover切换使任意从备节点升主
    7、在新的备集群首备节点内查看原始表数据
    8、在新的备集群从备节点查看原始表数据
    9、清理环境，恢复环境初始状态
Expect      :
    1、切换成功，原主集群降为备集群，原备集群升为主集群
    2、设置成功，重启成功
    3、创建成功，插入数据成功
    4、查看成功，与主集群内查看数据一致
    5、查看失败，从备节点不可读
    6、备集群内切换成功，从备升为首备，集群状态正常
    7、查看成功
    8、查看失败不可读
    9、清理成功，恢复环境成功
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonThread import CommonThread
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class DisasterRTOStandbyRead(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.p_comsh = [CommonSH(n) for n in ('P_PrimaryDbUser', 'P_Standby1DbUser', 'P_Standby2DbUser')]
        self.s_comsh = [CommonSH(n) for n in ('S_PrimaryDbUser', 'S_Standby1DbUser', 'S_Standby2DbUser')]
        self.user_node = [Node('P_PrimaryDbUser'), Node('S_PrimaryDbUser')]
        self.constant = Constant()

        self.param_list = ['recovery_parse_workers', 'recovery_redo_workers', 'ss_disaster_mode']
        self.value_list = ['4', '4']
        self.default_value = []
        self.table_list = ['ss_rto_012', 'ss_rto_012_new']
        self.pri_switch = f"gs_ddr -t switchover -m disaster_standby"
        self.sta_switch = f"gs_ddr -t switchover -m primary"

        self.log.info('---校验当前环境是否为网络复制双集群---')
        mode = self.p_comsh[0].execut_db_sql(f'show {self.param_list[2]};')
        self.log.info(mode)
        self.assertIn('stream', mode, '当前环境不是网络复制双集群')

    def test_rto_standby_read(self):
        text = '-----step1:原主备集群进行集群间切换，查看切换后的集群状态  expect:切换成功，原主集群降为备集群，原备集群升为主集群-----'
        self.log.info(text)

        self.log.info('---原主集群进行集群间切换---')
        pri_switch = CommonThread(self.user_node[0].sh, args=(self.pri_switch,))
        pri_switch.setDaemon(True)
        pri_switch.start()

        self.log.info('---原备集群进行集群间切换---')
        sta_switch = CommonThread(self.user_node[1].sh, args=(self.sta_switch,))
        sta_switch.setDaemon(True)
        sta_switch.start()

        pri_switch.join(10 * 60)
        ret1 = pri_switch.get_result()
        self.log.info(ret1)
        self.assertIn('Successfully do dorado disaster recovery switchover', ret1, '原主集群切换失败')

        sta_switch.join(10 * 60)
        ret2 = sta_switch.get_result()
        self.log.info(ret2)
        self.assertIn('Successfully do dorado disaster recovery switchover', ret2, '原备集群切换失败')

        self.log.info('---查看新主集群状态---')
        pri_status = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(pri_status)
        self.assertIn('cluster_status  : Normal', pri_status, '新主集群状态异常')
        self.assertIn('Primary Normal', pri_status.splitlines()[-1], '新主集群状态异常')

        self.log.info('---查看新备集群状态---')
        sta_status = self.p_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(sta_status)
        self.assertIn('cluster_status  : Normal', sta_status, '新备集群状态异常')
        self.assertIn('Main Standby Normal', sta_status.splitlines()[-1], '新备集群状态异常')

        text = '-----step2:新备集群内开启极致RTO特性，重启使其生效;  expect:设置成功，重启成功-----'
        self.log.info(text)

        for param in self.param_list[:2]:
            value = self.p_comsh[0].execut_db_sql(f'show {param}')
            self.default_value.append(value.splitlines()[-2].strip())
        self.log.info(self.default_value)

        for param in self.param_list[:2]:
            set_param = self.p_comsh[0].execute_gsguc('set', self.constant.GSGUC_SUCCESS_MSG,
                                                      f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(set_param)
            self.assertTrue(set_param, '备集群参数设置失败')

        self.log.info('---重启集群使参数生效---')
        restart = self.p_comsh[0].restart_db_cluster()
        self.log.info(restart)
        self.assertTrue(restart, '备集群集群重启失败')

        self.log.info('---查看参数值修改是否成功---')
        for param in self.param_list[:2]:
            pri_value = self.p_comsh[0].execut_db_sql(f'show {param};')
            self.assertEqual(pri_value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

        text = '-----step3:新主集群内创建行存段页式range分区表，插入数据  expect:创建成功，插入数据成功-----'
        self.log.info(text)

        sql1 = f"drop table if exists {self.table_list[0]};" \
               f"create table {self.table_list[0]}(id int, name varchar(200), age int, tel int) with(segment=on) " \
               f"partition by range(tel) " \
               f"(partition p1 values less than(5000), partition p2 values less than(10000), " \
               f"partition p3 values less than(15000), partition p4 values less than(20000), " \
               f"partition p5 values less than(25000), partition p6 values less than(30000), " \
               f"partition p7 values less than(40000), partition p8 values less than(50000)); " \
               f"declare    " \
               f"    i integer;" \
               f"begin" \
               f"    for i in 1..100000 loop" \
               f"    insert into {self.table_list[0]} values(random()*100,'name_'||random()*10,random()*100,random()*10000);" \
               f"    end loop;" \
               f"end;" \
               f"alter table {self.table_list[0]} rename to {self.table_list[1]};" \
               f"alter table {self.table_list[1]} add new_col1 int, add new_col2 text first, add new_col3 float after new_col2;" \
               f"update {self.table_list[1]} set tel = tel * 2 where id%2 != 0 and id < 1000;" \
               f"delete from {self.table_list[1]} where id < 20;"
        self.log.info(sql1)

        res1 = self.s_comsh[0].execut_db_sql(sql1)
        self.log.info(res1)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res1, '建表失败')
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG, res1, '匿名块执行失败，插入数据失败')
        self.assertIn(self.constant.ALTER_TABLE_MSG, res1, 'alter表失败')
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, res1, 'update表数据失败')
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, res1, 'delete表数据失败')

        text = '-----step4:新备集群首备节点查看表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待备集群数据同步---')
        flag = self.p_comsh[0].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        sql2 = f"select count(*) from {self.table_new};"
        self.log.info(sql2)
        select_res = self.p_comsh[0].execut_db_sql(sql2)
        self.log.info(select_res)

        p_select = self.s_comsh[0].execut_db_sql(sql2)
        self.log.info(p_select)
        self.assertEqual(p_select.splitlines()[-2].strip(), select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step5:新备集群从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.p_comsh[1].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

        text = '-----step6:新备集群内执行switchover切换使从备节点升主，查看集群状态  ' \
               'expect:备集群内节点切换成功，集群状态正常-----'
        self.log.info(text)

        self.log.info('---执行switchover切换---')
        switch = self.p_comsh[0].exec_cm_ctl('switchover', '-n 2', f'-D {macro.DB_INSTANCE_PATH}')
        self.log.info(switch)
        self.assertIn(self.constant.cm_switchover_success_msg, switch, '执行失败' + text)

        self.log.info('---查看切换后的集群状态---')
        status2 = self.p_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status2)
        self.assertIn('cluster_state  : Normal', status2, '集群状态异常')
        self.assertIn('Main Standby Normal', status2.splitlines()[-1].split('|')[1], '备集群内未切换成功')

        text = '-----step7:在新的首备节点内查看原始表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待新的首备节点数据同步---')
        flag = self.p_comsh[1].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        new_s_select = self.p_comsh[1].execut_db_sql(sql2)
        self.log.info(new_s_select)
        self.assertEqual(new_s_select.splitlines()[-2].strip(),
                         select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step8:在新的从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.p_comsh[0].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

    def tearDown(self):
        text = '-----step9:清理环境，恢复环境初始状态  expect:清理成功，恢复环境成功-----'
        self.log.info(text)

        self.log.info('-----清理表-----')
        drop = f"drop table {self.table_list[1]} cascade;"
        self.log.info(drop)
        drop_res = self.s_comsh[1].execut_db_sql(drop)
        self.log.info(drop_res)

        self.log.info('-----恢复环境默认参数-----')
        for param in self.param_list[:2]:
            reset_param = self.p_comsh[0].execute_gsguc('set',
                                                        self.constant.GSGUC_SUCCESS_MSG,
                                                        f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(reset_param)

        self.log.info('---新主集群执行switchover---')
        switch_exec3 = CommonThread(self.user_node[1], args=(self.sta_switch,))
        switch_exec3.setDaemon(True)
        switch_exec3.start()

        self.log.info('---新备集群执行switchover---')
        switch_exec4 = CommonThread(self.user_node[0], args=(self.pri_switch,))
        switch_exec4.setDaemon(True)
        switch_exec4.start()

        switch_exec3.join(10 * 60)
        ret3 = switch_exec3.get_result()
        self.log.info(ret3)

        switch_exec4.join(10 * 60)
        ret4 = switch_exec4.get_result()
        self.log.info(ret4)

        self.log.info('-----查看参数默认值是否恢复-----')
        for param in self.param_list[:2]:
            reset_value = self.s_comsh[0].execut_db_sql(f'show {param};')
            self.log.info(reset_value)
            self.assertEqual(reset_value.splitlines()[-2].strip(),
                             self.default_value[self.param_list.index(param)], '参数恢复失败')

        status = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status)

        time.sleep(60)
        self.log.info('-----恢复环境初始状态-----')
        switchover = self.s_comsh[0].exec_cm_ctl(mode='switchover', param='-a')
        self.log.info(switchover)

        self.assertIn('cluster_state  : Normal', status, '集群状态异常')
        self.assertIn('Main Standby Normal', status, '执行失败' + text)
        self.assertEqual(1, self.drop_res.count(self.constant.DROP_TABLE_SUCCESS), '删除表失败')
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret3, '主集群降备失败')
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret4, '备集群升主失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
